package R;

import U.Utils;
import rx.Notification;
import rx.Observable;
import rx.Observable.Operator;
import rx.Producer;
import rx.Subscriber;
import rx.functions.Action1;
import rx.functions.Func1;
import rx.internal.util.InternalObservableUtils;
import rx.subjects.BehaviorSubject;
import rx.subjects.Subject;

public class R {
    public static Observable<Integer> observable = Observable.from(Utils.INTS);

    public static boolean retried = false;
    private static Subscriber<Object> subscriber;

    public static void main(String[] args) {
//        repeat();
//          retry();
//        repeatWhen();
//        retryWhen();
//        subjectTest();

    }

    static boolean stopOnComplete = true;
    static boolean stopOnError = false;

    /**
     * 是否，我们接收到的参数thrwoable做出的行为，可以被拦截
     */
    private static void subjectTest() {

        Subject<Notification<?>, Notification<?>> terminals = BehaviorSubject.create();
        Func1<Observable<? extends Throwable>, Observable<?>> notificationHandler = new Func1<Observable<? extends Throwable>, Observable<?>>() {
            public Observable<?> call(Observable<? extends Throwable> err) {

                System.out.println(" handlerLogic: " + err);
                //通过map 或者其他操作符号获取到item
                return err.map(new Func1<Throwable, Notification>() {

                    public Notification call(Throwable throwable) {
                        System.out.println(throwable.getMessage() + " ---");
                        return Notification.createOnNext("嘿嘿");

                    }
                });

            }
        };

        Func1<Observable<? extends Notification<?>>, Observable<?>> realHandler = InternalObservableUtils.createRetryDematerializer(notificationHandler);


        Observable<Notification<?>> observable = terminals.lift(new Operator<Notification<?>, Notification<?>>() {

            public Subscriber<? super Notification<?>> call(final Subscriber<? super Notification<?>> filteredTerminals) {
                return new Subscriber<Notification<?>>(filteredTerminals) {

                    public void onCompleted() {
                        System.out.println("#terminal=>onCompleted");
                        filteredTerminals.onCompleted();
                    }

                    public void onError(Throwable e) {
                        System.out.println("#terminal=>onError:" + e.getMessage());
                        filteredTerminals.onError(e);
                    }

                    public void onNext(Notification<?> t) {
                        System.out.println("here 拦截" + t);
                        if (t.isOnCompleted() && stopOnComplete) {
                            filteredTerminals.onCompleted();
                        } else if (t.isOnError() && stopOnError) {
                            filteredTerminals.onError(t.getThrowable());
                        } else {
                            System.out.println("filteredTerminals: " + filteredTerminals);
                            filteredTerminals.onNext(t);
                        }
                    }

                    @Override
                    public void setProducer(Producer producer) {
                        producer.request(Long.MAX_VALUE);
                    }
                };
            }
        });
        Observable<Throwable> map = observable.map(new Func1<Notification<?>, Throwable>() {
            public Throwable call(Notification<?> notification) {
                return notification.getThrowable();
            }
        });
        Observable<String> observable1 = map.map(new Func1<Throwable, String>() {

            public String call(Throwable throwable) {
                System.out.println(throwable.getMessage() + " ---");
                return "heihi";

            }
        });
        subscriber = new Subscriber<Object>() {
            public void onCompleted() {
                System.out.println("onCompleted");

            }

            public void onError(Throwable e) {
                System.out.println("onError" + e.getMessage());

            }

            public void onNext(Object o) {
                System.out.println(subscriber);
                System.out.println("@@@ " + o);
            }
        };
        observable1.subscribe(subscriber);

//        terminals.onNext(Notification.createOnNext("Single"));
        terminals.onNext(Notification.createOnError(new Throwable("DoubleErrorrrrrrrrrrrrrr")));
    }

    /**
     * 一旦发生错误(onError)，就重新订阅
     */
    private static void retry() {
        Observable<Integer> observable = Observable.create(new Observable.OnSubscribe<Integer>() {
            public void call(Subscriber<? super Integer> subscriber) {
                for (int k = 0; k < 4; k++) {
                    if (k == 2)
                        subscriber.onError(new Throwable("index 2 illegal!"));
                    else subscriber.onNext(k);
                }
                subscriber.onCompleted();
            }
        });
        observable.retry().subscribe(new Action1<Integer>() {
            public void call(Integer integer) {
                System.out.println(integer);
            }
        });

        //最多重试多少次
        //observable.retry(2)

    }

    /**
     * retry()方法的局限在于无法自定义重试的逻辑，比如延迟一段时间再重新订阅(重试)，或者是发生特定的错误才重试，等等。
     * 所以retryWhen就是解决这个问题的.具体逻辑是，当source 发送了一个error时，会将这个error包装成observable传给
     * notificationHandler(notificationHandler函数本身是返回一个Observable，记作t)，如果t 发送了complete事件，那么
     * 就认作不想重试，结束订阅，不然的话就重新订阅source.所以当错误传来的时候我们可以根据具体的情况选择 返回一个发送completer事件
     * 的observable；或者是发送next或者error事件的observable来重新订阅。p.s. 源码内的注释指出是t发送complete或者error事件都可以
     * 停止重试，实际是只有complete事件可以中断重试。
     */
    private static int retryCount = 0;

    private static void retryWhen() {
        //只重试一次的逻辑
        Observable.create(new Observable.OnSubscribe<Integer>() {
            public void call(Subscriber<? super Integer> subscriber) {

                subscriber.onNext(3);
                subscriber.onNext(1 / 0);
                subscriber.onNext(10 / 2);
                subscriber.onCompleted();

            }
        }).retryWhen(new Func1<Observable<? extends Throwable>, Observable<?>>() {
            public Observable<?> call(Observable<? extends Throwable> err) {
                return Observable.just(10000);
                //通过map 或者其他操作符号获取到item
//                err.flatMap(new Func1<Throwable, Observable<?>>() {
//
//                    public Observable<?> call(Throwable throwable) {
//                        if (retryCount < 3) {
//                            System.out.println("retry ++");
//                            retryCount++;
//                            return Observable.just("");
//                        } else {
//                            System.out.println("no retry");
//                            return Observable.empty();
//                        }
//                    }
//                });
//                return err;

//                return Observable.error(new Throwable("done!"));


            }
        })
                .subscribe(new Subscriber<Integer>() {
                    public void onCompleted() {
                        System.out.println("onComplete==");
                    }

                    public void onError(Throwable e) {
                        System.out.println(e.getMessage() + "????");
                    }

                    public void onNext(Integer integer) {
                        System.out.println(integer + "==hi");

                    }
                });
    }

    /**
     * 一旦结束(onComplete)就重复
     */
    private static void repeat() {
        observable.repeat().subscribe(new Action1<Integer>() {
            public void call(Integer integer) {
                System.out.println(integer);
            }
        });
    }

    /**
     * 与retryWhen类似
     */
    private static void repeatWhen() {
        observable.repeatWhen(new Func1<Observable<? extends Void>, Observable<?>>() {
            public Observable<?> call(Observable<? extends Void> observable) {
                return observable.isEmpty();
            }
        }).onBackpressureDrop();
    }

}
